package org.hadoop.spark
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
object UpdateStateByKeyClient {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("NetworkWordCount")
    //声明SparkStreaming对象
    val ssc = new StreamingContext(conf, Seconds(1));
    //必须要设置检查点
    ssc.checkpoint("file:///D:/a/cp");
    //声明updateStateByKey函数=(参数){函数体}
    /** 其中第一个参数，为前面累计的结果，第二个参数为新的结果 * */
    val updateStateByKey = (values: Seq[Int], state: Option[Int]) => {
      val newValue: Int = state.getOrElse(0);
      //如果没有则为0
      //将前面的值进行累计求和
      val sum: Int = values.sum;
      Some(sum + newValue); //这是返回值
    };
    val rdd = ssc.socketTextStream("10.69.21.206", 9999);
    //转换完成以后再执行update函数
    val rdd2 = rdd.flatMap(_.split("\\s+")).map((_, 1))
      .updateStateByKey[Int](updateStateByKey);
    rdd2.print();
    ssc.start();
    ssc.awaitTermination();
    Thread.sleep(1000 * 30);
    ssc.stop();
  }
}
